#pragma once
#include "Headerfile.hpp"
#include "buffer.hpp"
namespace shw
{
    using fcnt = std::function<void(Buffer &)>;
    enum class AsyncLooperType
    {
        AsyncLooper_SAFE=0,
        AsyncLooper_UNSAFE,
    };
    class AsyncLooper
    {
    public:
        using ptr = std::shared_ptr<AsyncLooper>;
        AsyncLooper(const fcnt &fcn, AsyncLooperType type = AsyncLooperType::AsyncLooper_SAFE)
            : _type(type),_stopFlag(true), _fcn(fcn), _thread(std::thread(&AsyncLooper::ThreadRoute, this)) {}
        ~AsyncLooper() { Stop(); }
        void Stop() // 停止异步工作器
        {
            _stopFlag = false;
            _con_cum.notify_all(); // 唤醒所有消费线程
            _thread.join();        // 回收线程
        }
        void Push(const std::string &msg) // 生产数据
        {
            if (_stopFlag == false)
                return;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                if (_type == AsyncLooperType::AsyncLooper_SAFE)
                {
                    _con_pro.wait(lock, [&]()
                                  { return _proBuffer.WriteAbleSize() >= msg.size(); }); // 为真时往下走
                    
                }
                _proBuffer.Push(msg.c_str(), msg.size());
            }
            _con_cum.notify_all(); // 唤醒消费者处理数据
        }

    public:
        void ThreadRoute() // 消费数据
        {

            while (1)
            {
                {
                    if (_stopFlag == false && _proBuffer.Empty())
                        return;
                    std::unique_lock<std::mutex> lock(_mutex);
                    _con_cum.wait(lock, [&]()
                                  { return !_proBuffer.Empty() || _stopFlag == false; }); // 交换
                    if (_cumBuffer.Empty())  // 空了才交换：消费者缓冲区空了才和生产者交换
                    {
                        _cumBuffer.Swap(_proBuffer);
                    }
                }
                _con_pro.notify_all();
                _fcn(_cumBuffer);//从读指针位置开始处理
                _cumBuffer.Reset();
            }
        }
        Buffer GetproBuffer() { return _proBuffer; }

    private:
        AsyncLooperType _type;
        std::atomic<bool> _stopFlag;      // 异步工作器停止标志，为真停止
        std::mutex _mutex;                // 互斥锁
        Buffer _proBuffer;                // 生产缓冲区
        Buffer _cumBuffer;                // 消费缓冲区
        std::condition_variable _con_pro; // 生产者条件变量
        std::condition_variable _con_cum; // 消费者条件变量
        fcnt _fcn;                        // 处理消费者缓冲区数据的回调函数
        std::thread _thread;              // 异步工作线程
    };
}